1 /*
2 * Copyright (C) 2007 The Guava Authors
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 package com.google.common.collect;
18
19 import static com.google.common.base.Preconditions.checkArgument;
20 import static com.google.common.base.Preconditions.checkNotNull;
21 import static com.google.common.collect.CollectPreconditions.checkNonnegative;
22 import static com.google.common.collect.CollectPreconditions.checkRemove;
23
24 import com.google.common.annotations.Beta;
25 import com.google.common.annotations.VisibleForTesting;
26 import com.google.common.collect.Serialization.FieldSetter;
27 import com.google.common.math.IntMath;
28 import com.google.common.primitives.Ints;
29
30 import java.io.IOException;
31 import java.io.ObjectInputStream;
32 import java.io.ObjectOutputStream;
33 import java.io.Serializable;
34 import java.util.Collection;
35 import java.util.Iterator;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Set;
39 import java.util.concurrent.ConcurrentHashMap;
40 import java.util.concurrent.ConcurrentMap;
41 import java.util.concurrent.atomic.AtomicInteger;
42
43 import javax.annotation.Nullable;
44
45 /**
46 * A multiset that supports concurrent modifications and that provides atomic versions of most
47 * {@code Multiset} operations (exceptions where noted). Null elements are not supported.
48 *
49 * <p>See the Guava User Guide article on <a href=
50 * "http://code.google.com/p/guava-libraries/wiki/NewCollectionTypesExplained#Multiset">
51 * {@code Multiset}</a>.
52 *
53 * @author Cliff L. Biffle
54 * @author mike nonemacher
55 * @since 2.0 (imported from Google Collections Library)
56 */
57 public final class ConcurrentHashMultiset<E> extends AbstractMultiset<E> implements Serializable {
58
59 /*
60 * The ConcurrentHashMultiset's atomic operations are implemented primarily in terms of
61 * AtomicInteger's atomic operations, with some help from ConcurrentMap's atomic operations on
62 * creation and removal (including automatic removal of zeroes). If the modification of an
63 * AtomicInteger results in zero, we compareAndSet the value to zero; if that succeeds, we remove
64 * the entry from the Map. If another operation sees a zero in the map, it knows that the entry is
65 * about to be removed, so this operation may remove it (often by replacing it with a new
66 * AtomicInteger).
67 */
68
69 /** The number of occurrences of each element. */
70 private final transient ConcurrentMap<E, AtomicInteger> countMap;
71
72 // This constant allows the deserialization code to set a final field. This holder class
73 // makes sure it is not initialized unless an instance is deserialized.
74 private static class FieldSettersHolder {
75 static final FieldSetter<ConcurrentHashMultiset> COUNT_MAP_FIELD_SETTER =
76 Serialization.getFieldSetter(ConcurrentHashMultiset.class, "countMap");
77 }
78
79 /**
80 * Creates a new, empty {@code ConcurrentHashMultiset} using the default
81 * initial capacity, load factor, and concurrency settings.
82 */
83 public static <E> ConcurrentHashMultiset<E> create() {
84 // TODO(schmoe): provide a way to use this class with other (possibly arbitrary)
85 // ConcurrentMap implementors. One possibility is to extract most of this class into
86 // an AbstractConcurrentMapMultiset.
87 return new ConcurrentHashMultiset<E>(new ConcurrentHashMap<E, AtomicInteger>());
88 }
89
90 /**
91 * Creates a new {@code ConcurrentHashMultiset} containing the specified elements, using
92 * the default initial capacity, load factor, and concurrency settings.
93 *
94 * <p>This implementation is highly efficient when {@code elements} is itself a {@link Multiset}.
95 *
96 * @param elements the elements that the multiset should contain
97 */
98 public static <E> ConcurrentHashMultiset<E> create(Iterable<? extends E> elements) {
99 ConcurrentHashMultiset<E> multiset = ConcurrentHashMultiset.create();
100 Iterables.addAll(multiset, elements);
101 return multiset;
102 }
103
104 /**
105 * Creates a new, empty {@code ConcurrentHashMultiset} using {@code mapMaker}
106 * to construct the internal backing map.
107 *
108 * <p>If this {@link MapMaker} is configured to use entry eviction of any kind, this eviction
109 * applies to all occurrences of a given element as a single unit. However, most updates to the
110 * multiset do not count as map updates at all, since we're usually just mutating the value
111 * stored in the map, so {@link MapMaker#expireAfterAccess} makes sense (evict the entry that
112 * was queried or updated longest ago), but {@link MapMaker#expireAfterWrite} doesn't, because
113 * the eviction time is measured from when we saw the first occurrence of the object.
114 *
115 * <p>The returned multiset is serializable but any serialization caveats
116 * given in {@code MapMaker} apply.
117 *
118 * <p>Finally, soft/weak values can be used but are not very useful: the values are created
119 * internally and not exposed externally, so no one else will have a strong reference to the
120 * values. Weak keys on the other hand can be useful in some scenarios.
121 *
122 * @since 15.0 (source compatible (accepting the since removed {@code GenericMapMaker} class)
123 * since 7.0)
124 */
125 @Beta
126 public static <E> ConcurrentHashMultiset<E> create(MapMaker mapMaker) {
127 return new ConcurrentHashMultiset<E>(mapMaker.<E, AtomicInteger>makeMap());
128 }
129
130 /**
131 * Creates an instance using {@code countMap} to store elements and their counts.
132 *
133 * <p>This instance will assume ownership of {@code countMap}, and other code
134 * should not maintain references to the map or modify it in any way.
135 *
136 * @param countMap backing map for storing the elements in the multiset and
137 * their counts. It must be empty.
138 * @throws IllegalArgumentException if {@code countMap} is not empty
139 */
140 @VisibleForTesting ConcurrentHashMultiset(ConcurrentMap<E, AtomicInteger> countMap) {
141 checkArgument(countMap.isEmpty());
142 this.countMap = countMap;
143 }
144
145 // Query Operations
146
147 /**
148 * Returns the number of occurrences of {@code element} in this multiset.
149 *
150 * @param element the element to look for
151 * @return the nonnegative number of occurrences of the element
152 */
153 @Override public int count(@Nullable Object element) {
154 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
155 return (existingCounter == null) ? 0 : existingCounter.get();
156 }
157
158 /**
159 * {@inheritDoc}
160 *
161 * <p>If the data in the multiset is modified by any other threads during this method,
162 * it is undefined which (if any) of these modifications will be reflected in the result.
163 */
164 @Override public int size() {
165 long sum = 0L;
166 for (AtomicInteger value : countMap.values()) {
167 sum += value.get();
168 }
169 return Ints.saturatedCast(sum);
170 }
171
172 /*
173 * Note: the superclass toArray() methods assume that size() gives a correct
174 * answer, which ours does not.
175 */
176
177 @Override public Object[] toArray() {
178 return snapshot().toArray();
179 }
180
181 @Override public <T> T[] toArray(T[] array) {
182 return snapshot().toArray(array);
183 }
184
185 /*
186 * We'd love to use 'new ArrayList(this)' or 'list.addAll(this)', but
187 * either of these would recurse back to us again!
188 */
189 private List<E> snapshot() {
190 List<E> list = Lists.newArrayListWithExpectedSize(size());
191 for (Multiset.Entry<E> entry : entrySet()) {
192 E element = entry.getElement();
193 for (int i = entry.getCount(); i > 0; i--) {
194 list.add(element);
195 }
196 }
197 return list;
198 }
199
200 // Modification Operations
201
202 /**
203 * Adds a number of occurrences of the specified element to this multiset.
204 *
205 * @param element the element to add
206 * @param occurrences the number of occurrences to add
207 * @return the previous count of the element before the operation; possibly zero
208 * @throws IllegalArgumentException if {@code occurrences} is negative, or if
209 * the resulting amount would exceed {@link Integer#MAX_VALUE}
210 */
211 @Override public int add(E element, int occurrences) {
212 checkNotNull(element);
213 if (occurrences == 0) {
214 return count(element);
215 }
216 checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
217
218 while (true) {
219 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
220 if (existingCounter == null) {
221 existingCounter = countMap.putIfAbsent(element, new AtomicInteger(occurrences));
222 if (existingCounter == null) {
223 return 0;
224 }
225 // existingCounter != null: fall through to operate against the existing AtomicInteger
226 }
227
228 while (true) {
229 int oldValue = existingCounter.get();
230 if (oldValue != 0) {
231 try {
232 int newValue = IntMath.checkedAdd(oldValue, occurrences);
233 if (existingCounter.compareAndSet(oldValue, newValue)) {
234 // newValue can't == 0, so no need to check & remove
235 return oldValue;
236 }
237 } catch (ArithmeticException overflow) {
238 throw new IllegalArgumentException("Overflow adding " + occurrences
239 + " occurrences to a count of " + oldValue);
240 }
241 } else {
242 // In the case of a concurrent remove, we might observe a zero value, which means another
243 // thread is about to remove (element, existingCounter) from the map. Rather than wait,
244 // we can just do that work here.
245 AtomicInteger newCounter = new AtomicInteger(occurrences);
246 if ((countMap.putIfAbsent(element, newCounter) == null)
247 || countMap.replace(element, existingCounter, newCounter)) {
248 return 0;
249 }
250 break;
251 }
252 }
253
254 // If we're still here, there was a race, so just try again.
255 }
256 }
257
258 /**
259 * Removes a number of occurrences of the specified element from this multiset. If the multiset
260 * contains fewer than this number of occurrences to begin with, all occurrences will be removed.
261 *
262 * @param element the element whose occurrences should be removed
263 * @param occurrences the number of occurrences of the element to remove
264 * @return the count of the element before the operation; possibly zero
265 * @throws IllegalArgumentException if {@code occurrences} is negative
266 */
267 /*
268 * TODO(cpovirk): remove and removeExactly currently accept null inputs only
269 * if occurrences == 0. This satisfies both NullPointerTester and
270 * CollectionRemoveTester.testRemove_nullAllowed, but it's not clear that it's
271 * a good policy, especially because, in order for the test to pass, the
272 * parameter must be misleadingly annotated as @Nullable. I suspect that
273 * we'll want to remove @Nullable, add an eager checkNotNull, and loosen up
274 * testRemove_nullAllowed.
275 */
276 @Override public int remove(@Nullable Object element, int occurrences) {
277 if (occurrences == 0) {
278 return count(element);
279 }
280 checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
281
282 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
283 if (existingCounter == null) {
284 return 0;
285 }
286 while (true) {
287 int oldValue = existingCounter.get();
288 if (oldValue != 0) {
289 int newValue = Math.max(0, oldValue - occurrences);
290 if (existingCounter.compareAndSet(oldValue, newValue)) {
291 if (newValue == 0) {
292 // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
293 // another thread has already replaced it with a new counter, which is fine.
294 countMap.remove(element, existingCounter);
295 }
296 return oldValue;
297 }
298 } else {
299 return 0;
300 }
301 }
302 }
303
304 /**
305 * Removes exactly the specified number of occurrences of {@code element}, or makes no
306 * change if this is not possible.
307 *
308 * <p>This method, in contrast to {@link #remove(Object, int)}, has no effect when the
309 * element count is smaller than {@code occurrences}.
310 *
311 * @param element the element to remove
312 * @param occurrences the number of occurrences of {@code element} to remove
313 * @return {@code true} if the removal was possible (including if {@code occurrences} is zero)
314 */
315 public boolean removeExactly(@Nullable Object element, int occurrences) {
316 if (occurrences == 0) {
317 return true;
318 }
319 checkArgument(occurrences > 0, "Invalid occurrences: %s", occurrences);
320
321 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
322 if (existingCounter == null) {
323 return false;
324 }
325 while (true) {
326 int oldValue = existingCounter.get();
327 if (oldValue < occurrences) {
328 return false;
329 }
330 int newValue = oldValue - occurrences;
331 if (existingCounter.compareAndSet(oldValue, newValue)) {
332 if (newValue == 0) {
333 // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
334 // another thread has already replaced it with a new counter, which is fine.
335 countMap.remove(element, existingCounter);
336 }
337 return true;
338 }
339 }
340 }
341
342 /**
343 * Adds or removes occurrences of {@code element} such that the {@link #count} of the
344 * element becomes {@code count}.
345 *
346 * @return the count of {@code element} in the multiset before this call
347 * @throws IllegalArgumentException if {@code count} is negative
348 */
349 @Override public int setCount(E element, int count) {
350 checkNotNull(element);
351 checkNonnegative(count, "count");
352 while (true) {
353 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
354 if (existingCounter == null) {
355 if (count == 0) {
356 return 0;
357 } else {
358 existingCounter = countMap.putIfAbsent(element, new AtomicInteger(count));
359 if (existingCounter == null) {
360 return 0;
361 }
362 // existingCounter != null: fall through
363 }
364 }
365
366 while (true) {
367 int oldValue = existingCounter.get();
368 if (oldValue == 0) {
369 if (count == 0) {
370 return 0;
371 } else {
372 AtomicInteger newCounter = new AtomicInteger(count);
373 if ((countMap.putIfAbsent(element, newCounter) == null)
374 || countMap.replace(element, existingCounter, newCounter)) {
375 return 0;
376 }
377 }
378 break;
379 } else {
380 if (existingCounter.compareAndSet(oldValue, count)) {
381 if (count == 0) {
382 // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
383 // another thread has already replaced it with a new counter, which is fine.
384 countMap.remove(element, existingCounter);
385 }
386 return oldValue;
387 }
388 }
389 }
390 }
391 }
392
393 /**
394 * Sets the number of occurrences of {@code element} to {@code newCount}, but only if
395 * the count is currently {@code expectedOldCount}. If {@code element} does not appear
396 * in the multiset exactly {@code expectedOldCount} times, no changes will be made.
397 *
398 * @return {@code true} if the change was successful. This usually indicates
399 * that the multiset has been modified, but not always: in the case that
400 * {@code expectedOldCount == newCount}, the method will return {@code true} if
401 * the condition was met.
402 * @throws IllegalArgumentException if {@code expectedOldCount} or {@code newCount} is negative
403 */
404 @Override public boolean setCount(E element, int expectedOldCount, int newCount) {
405 checkNotNull(element);
406 checkNonnegative(expectedOldCount, "oldCount");
407 checkNonnegative(newCount, "newCount");
408
409 AtomicInteger existingCounter = Maps.safeGet(countMap, element);
410 if (existingCounter == null) {
411 if (expectedOldCount != 0) {
412 return false;
413 } else if (newCount == 0) {
414 return true;
415 } else {
416 // if our write lost the race, it must have lost to a nonzero value, so we can stop
417 return countMap.putIfAbsent(element, new AtomicInteger(newCount)) == null;
418 }
419 }
420 int oldValue = existingCounter.get();
421 if (oldValue == expectedOldCount) {
422 if (oldValue == 0) {
423 if (newCount == 0) {
424 // Just observed a 0; try to remove the entry to clean up the map
425 countMap.remove(element, existingCounter);
426 return true;
427 } else {
428 AtomicInteger newCounter = new AtomicInteger(newCount);
429 return (countMap.putIfAbsent(element, newCounter) == null)
430 || countMap.replace(element, existingCounter, newCounter);
431 }
432 } else {
433 if (existingCounter.compareAndSet(oldValue, newCount)) {
434 if (newCount == 0) {
435 // Just CASed to 0; remove the entry to clean up the map. If the removal fails,
436 // another thread has already replaced it with a new counter, which is fine.
437 countMap.remove(element, existingCounter);
438 }
439 return true;
440 }
441 }
442 }
443 return false;
444 }
445
446 // Views
447
448 @Override Set<E> createElementSet() {
449 final Set<E> delegate = countMap.keySet();
450 return new ForwardingSet<E>() {
451 @Override protected Set<E> delegate() {
452 return delegate;
453 }
454
455 @Override
456 public boolean contains(@Nullable Object object) {
457 return object != null && Collections2.safeContains(delegate, object);
458 }
459
460 @Override
461 public boolean containsAll(Collection<?> collection) {
462 return standardContainsAll(collection);
463 }
464
465 @Override public boolean remove(Object object) {
466 return object != null && Collections2.safeRemove(delegate, object);
467 }
468
469 @Override public boolean removeAll(Collection<?> c) {
470 return standardRemoveAll(c);
471 }
472 };
473 }
474
475 @Override public Set<Multiset.Entry<E>> createEntrySet() {
476 return new EntrySet();
477 }
478
479 @Override int distinctElements() {
480 return countMap.size();
481 }
482
483 @Override public boolean isEmpty() {
484 return countMap.isEmpty();
485 }
486
487 @Override Iterator<Entry<E>> entryIterator() {
488 // AbstractIterator makes this fairly clean, but it doesn't support remove(). To support
489 // remove(), we create an AbstractIterator, and then use ForwardingIterator to delegate to it.
490 final Iterator<Entry<E>> readOnlyIterator =
491 new AbstractIterator<Entry<E>>() {
492 private Iterator<Map.Entry<E, AtomicInteger>> mapEntries = countMap.entrySet().iterator();
493
494 @Override protected Entry<E> computeNext() {
495 while (true) {
496 if (!mapEntries.hasNext()) {
497 return endOfData();
498 }
499 Map.Entry<E, AtomicInteger> mapEntry = mapEntries.next();
500 int count = mapEntry.getValue().get();
501 if (count != 0) {
502 return Multisets.immutableEntry(mapEntry.getKey(), count);
503 }
504 }
505 }
506 };
507
508 return new ForwardingIterator<Entry<E>>() {
509 private Entry<E> last;
510
511 @Override protected Iterator<Entry<E>> delegate() {
512 return readOnlyIterator;
513 }
514
515 @Override public Entry<E> next() {
516 last = super.next();
517 return last;
518 }
519
520 @Override public void remove() {
521 checkRemove(last != null);
522 ConcurrentHashMultiset.this.setCount(last.getElement(), 0);
523 last = null;
524 }
525 };
526 }
527
528 @Override public void clear() {
529 countMap.clear();
530 }
531
532 private class EntrySet extends AbstractMultiset<E>.EntrySet {
533 @Override ConcurrentHashMultiset<E> multiset() {
534 return ConcurrentHashMultiset.this;
535 }
536
537 /*
538 * Note: the superclass toArray() methods assume that size() gives a correct
539 * answer, which ours does not.
540 */
541
542 @Override public Object[] toArray() {
543 return snapshot().toArray();
544 }
545
546 @Override public <T> T[] toArray(T[] array) {
547 return snapshot().toArray(array);
548 }
549
550 private List<Multiset.Entry<E>> snapshot() {
551 List<Multiset.Entry<E>> list = Lists.newArrayListWithExpectedSize(size());
552 // Not Iterables.addAll(list, this), because that'll forward right back here.
553 Iterators.addAll(list, iterator());
554 return list;
555 }
556 }
557
558 /**
559 * @serialData the ConcurrentMap of elements and their counts.
560 */
561 private void writeObject(ObjectOutputStream stream) throws IOException {
562 stream.defaultWriteObject();
563 stream.writeObject(countMap);
564 }
565
566 private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
567 stream.defaultReadObject();
568 @SuppressWarnings("unchecked") // reading data stored by writeObject
569 ConcurrentMap<E, Integer> deserializedCountMap =
570 (ConcurrentMap<E, Integer>) stream.readObject();
571 FieldSettersHolder.COUNT_MAP_FIELD_SETTER.set(this, deserializedCountMap);
572 }
573
574 private static final long serialVersionUID = 1;
575 }